package com.rv.nats.core;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rv.common.core.domain.R;
import com.rv.nats.annotation.NatsListener;
import com.rv.nats.annotation.Subscribe;
import com.rv.nats.enums.ClientType;
import com.rv.nats.enums.Mode;
import com.rv.nats.listener.NatsConnectionListener;
import com.rv.nats.listener.NatsErrorListener;
import com.rv.nats.resolver.param.ParameterResolver;
import com.rv.nats.resolver.param.ParameterResolverFactory;
import io.nats.client.*;
import io.nats.client.api.*;
import io.nats.client.impl.NatsMessage;
import lombok.extern.slf4j.Slf4j;
import org.reflections.Reflections;
import org.reflections.util.ClasspathHelper;
import org.reflections.util.ConfigurationBuilder;
import org.springframework.beans.factory.DisposableBean;

import java.io.IOException;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @time: 2022/9/20
 * @author: .sun
 */

@Slf4j
public abstract class AbstractNatsHandler implements NatsHandler, DisposableBean {

    /**
     * 客户端类型
     */
    protected ClientType clientType;

    /**
     * nats 连接
     */
    protected Connection nc;

    /**
     * nats Dispatcher
     */
    protected Dispatcher dispatcher;

    /**
     * nats JetStream 用于Stream持久化消息
     */
    protected JetStream js;

    /**
     * 消费者
     */
    private final Map<String, Method> consumerMap = new HashMap<>();

    /**
     * 订阅列表
     */
    private final Map<String, Subscription> subsList = new HashMap<>();


    /**
     * 异常监听器
     */
    protected ErrorListener errorListener = new NatsErrorListener();

    /**
     * 连接监听器
     */
    protected ConnectionListener connectionListener = new NatsConnectionListener();

    /**
     * 断开连接超时时间
     */
    protected Duration drainTimeout = Duration.ofSeconds(20);

    /**
     * 请求回复超时时间
     */
    protected Duration requestTimeout = Duration.ofSeconds(20);


    /**
     * json转换映射
     */
    private ObjectMapper mapper;


    /**
     * 需要扫描的包名
     * 默认为当前包下所有包
     */
    private String packageName = "";


    public AbstractNatsHandler() {
        this.mapper = new ObjectMapper();
        this.mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        this.packageName = getPackageName();
    }


    /**
     * 获取当前包名
     * @return 包名
     */
    protected String getPackageName() {
        this.packageName = this.getClass().getPackage().getName();
        int i = this.packageName.indexOf(".");
        return this.packageName.substring(0, i);
    }


    /**
     * 初始化
     *
     * @param builder            nats 配置构造
     * @param reconnectOnConnect 是否在连接时自动重连
     */
    public void init(Options.Builder builder, boolean reconnectOnConnect) throws IOException, InterruptedException {
        builder.connectionName(clientType.name());
        builder.errorListener(errorListener);
        builder.connectionListener(connectionListener);
        this.nc = reconnectOnConnect ? Nats.connectReconnectOnConnect(builder.build()) : Nats.connect(builder.build());
        this.dispatcher = this.nc.createDispatcher();
        this.js = nc.jetStream();
        subscribe();
    }

    /**
     * 用于nats初始化连接
     *
     * @param host 主机
     * @param port 端口号
     */
    @Override
    public void init(String host, Integer port) throws IOException, InterruptedException {
        Options.Builder builder = Options.builder();
        builder.server(host + ":" + port);
        this.init(builder, false);
    }


    /**
     * 用于nats初始化连接
     *
     * @param host     主机
     * @param port     端口号
     * @param username 用户名
     * @param password 密码
     */
    @Override
    public void init(String host, Integer port, String username, String password) throws IOException, InterruptedException {
        Options.Builder builder = Options.builder();
        builder.server(host + ":" + port);
        builder.userInfo(username, password);
        this.init(builder, false);
    }

    /**
     * 判断是否是连接的状态
     *
     * @return true:是 false:否
     */
    @Override
    public boolean isActive() {
        return nc != null && nc.getStatus() == Connection.Status.CONNECTED;
    }

    /**
     * @param topic 取消订阅的主题
     */
    @Override
    public void unsubscribe(String topic) {
        Subscription subscription = subsList.remove(topic);
        if (subscription != null && isActive() && dispatcher.isActive()) {
            dispatcher.unsubscribe(subscription);
        }
    }

    @Override
    public void publish(String subject, Object data) {
        String dataStr = JSON.toJSONString(data);
        Message message = new NatsMessage.Builder().subject(subject).data(dataStr).build();
        publish(message);
    }

    /**
     * 带回复主题的发布方法
     *
     * @param topic      发布主题
     * @param replyTopic 回复主题 , 这个需要发布者自己订阅
     * @param data       发布数据
     */
    @Override
    public void publish(String topic, String replyTopic, Object data) {
        Message message = new NatsMessage.Builder().subject(topic).replyTo(replyTopic).data(JSON.toJSONString(data)).build();
        publish(message);
    }

    @Override
    public void publish(String subject, byte[] data) {
        Message message = new NatsMessage.Builder().subject(subject).data(data).build();
        publish(message);
    }

    public void publish(Message message) {
        try {
            nc.publish(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过使用JetStream发布消息
     *
     * @param topic 发布的主题
     * @param data  发布的数据
     */
    @Override
    public PublishAck publishWithJs(String topic, Object data) {
        Message message = new NatsMessage.Builder().subject(topic).data(JSON.toJSONString(data)).build();
        try {
            return js.publish(message);
        } catch (Exception e) {
            log.error("[{}] request , subject:{} , exception", clientType, topic, e);
        }
        return null;
    }

    /**
     * @param topic      主题
     * @param payload    数据
     * @param returnType 返回数据的Class类型
     * @return 返回数据
     */
    public <T> T request(String topic, Object payload, JavaType returnType) {
        return request(topic, payload, requestTimeout, returnType);
    }


    public <T> T request(String subject, Object payload, Duration duration, JavaType returnType) {
        Message message;
        try{
            log.info("[{}] request , subject:{}  parameters:{}", clientType, subject, JSON.toJSONString(payload));
            long ct = System.currentTimeMillis();
            message = nc.request(subject, JSON.toJSONBytes(payload), duration);
            if (message == null) {
                log.warn("[{}] request finish , message is null , subject:{} , consume time(ms):{}", clientType, subject, System.currentTimeMillis() - ct);
                return null;
            }
            String result = new String(message.getData());
            log.info("[{}] request finish , subject:{} , consume time(ms):{}", clientType, subject, System.currentTimeMillis() - ct);
            return mapper.readValue(result, returnType);
        } catch (Exception e) {
            log.error("[{}] request error , subject:{} , exception", clientType, subject, e);
        }
        return null;
    }


    public <T> T request(String subject, Object payload, Duration duration, TypeReference<T> clazz) {
        Message message;
        try {

            log.info("client type [{}] start nats request , subject:{}  parameters:{}", clientType, subject, payload);
            long ct = System.currentTimeMillis();
            message = nc.request(subject, JSON.toJSONBytes(payload), duration);
            if (message == null) {
                return null;
            }
            String result = new String(message.getData());
            log.info("client type [{}] nats request end ,  subject:{} , consume time(ms):{}", clientType, subject, System.currentTimeMillis() - ct);
            return JSONObject.parseObject(result, clazz);
        } catch (Exception e) {
            log.error("[{}] request , subject:{} , exception", clientType, subject, e);
        }
        return null;
    }

    /**
     * @param topic 主题
     * @param data  数据
     * @param clazz 返回数据的Class类型
     */
    @Override
    public <T> T request(String topic, Object data, TypeReference<T> clazz) {
        return request(topic, data, requestTimeout, clazz);
    }


    /**
     * 通过使用JetStream发布消息
     *
     * @param message
     */
    @Override
    public PublishAck publishWithJs(Message message) {
        try {
            return js.publish(message);
        } catch (IOException | JetStreamApiException e) {
            throw new RuntimeException(e);
        }
    }



    private void subscribe(Method method , Object bean, Dispatcher dispatcher) {
        Subscribe subscribe = method.getAnnotation(Subscribe.class);
        String topic = subscribe.topic();
        String queue = subscribe.queue();
        MessageHandler handler = (message) -> handleMessage(method, bean, message);
        Subscription subscription = !queue.isEmpty() ? dispatcher.subscribe(topic, queue, handler) : dispatcher.subscribe(topic, handler);
        subsList.put(topic, subscription);
    }

    /**
     * push消费模式
     *
     * @param method 反射类的订阅的方法
     * @param bean   反射类
     */
    private void jetStreamSubscribe(Method method, Object bean) {
        if (js == null) {
            log.error("not support JetStream configuration");
            return;
        }
        Subscribe subscribe = method.getDeclaredAnnotation(Subscribe.class);
        ConsumerConfiguration configuration = configuration(method);
        String streamName = subscribe.streamName();
        try {
            if (subscribe.mode() == Mode.PULL) {
                PullSubscribeOptions.Builder builder = new PullSubscribeOptions.Builder();
                PullSubscribeOptions options = builder.stream(streamName).configuration(configuration).build();
                JetStreamSubscription subscription = js.subscribe(subscribe.topic(), options);
                nc.getOptions().getExecutor().execute(() -> {
                    while (isActive()) {
                        List<Message> messages = subscription.fetch(1, subscribe.ackWait());
                        messages.forEach((msg) -> handleMessage(method, bean, msg));
                    }
                });
            } else if (subscribe.mode() == Mode.PUSH) {
                PushSubscribeOptions.Builder builder = new PushSubscribeOptions.Builder();
                PushSubscribeOptions options = builder.stream(streamName).configuration(configuration).build();
                js.subscribe(subscribe.topic(), dispatcher, (message) -> handleMessage(method, bean, message), subscribe.autoAck(), options);
            }
        } catch (IOException | JetStreamApiException e) {
            throw new RuntimeException(e);
        }
    }

    private ConsumerConfiguration configuration(Method method) {

        Subscribe subscribe = method.getDeclaredAnnotation(Subscribe.class);
        Mode mode = subscribe.mode();
        String durable = String.format("@Consumer@%s#%s", mode, NUID.nextGlobal());
        consumerMap.put(durable, method);
        return new ConsumerConfiguration.Builder().deliverPolicy(DeliverPolicy.All).ackPolicy(AckPolicy.Explicit).ackWait(Duration.ofSeconds(subscribe.ackWait())).durable(durable).replayPolicy(ReplayPolicy.Instant).build();
    }

    protected void handleMessage(final Method method, final Object bean, Message message) {
        Subscribe subscribe = method.getAnnotation(Subscribe.class);
        String topic = subscribe.topic();
        try {
            boolean isLog = subscribe.log();
            if (isLog) {
                String response = new String(message.getData(), StandardCharsets.UTF_8);
                if (subscribe.mode() == Mode.SIMPLE) {
                    log.info("subject: {} ,  received data: {}", message.getSubject(), response);
                } else {
                    log.info("subject: {} , message seq: {} received data: {}", message.getSubject(), message.metaData().streamSequence(), response);
                }
            }
            Parameter[] parameters = method.getParameters();
            Object result = null;
            if (parameters.length == 0) {
                result = method.invoke(bean);
            }else{
                List<Object> args = new ArrayList<>();
                for (Parameter parameter : parameters) {
                    ParameterResolver resolver = ParameterResolverFactory.get(parameter);
                    args.add(resolver.resolve(message, parameter));
                }
                result = method.invoke(bean, args.toArray());
            }
            if (subscribe.reply() && message.getReplyTo() != null) {
                publish(message.getReplyTo(), result);
            }
            if (isLog) {
                log.info("reply subject:{} , result:[{}]", message.getReplyTo() != null ? message.getReplyTo() : "no reply subject", result);
            }
            if (message.isJetStream() && subscribe.autoAck()) {
                message.ack();
            }
        } catch (Exception e) {
            if (message.getReplyTo() != null) {
                publish(message.getReplyTo(), R.error(e.getMessage()));
            }
            log.error("nats report appear error , method : {} , topic:{}", method.getDeclaringClass().getName() + "." + method.getName(), topic, e);
        }
    }


    protected void subscribe() {
        log.info("[{}] start subscribe topic", this.getClientType());
        Reflections ref = new Reflections(new ConfigurationBuilder().setUrls(ClasspathHelper.forPackage(this.packageName)));
        Set<Class<?>> clazzs = ref.getTypesAnnotatedWith(NatsListener.class);
        for (Class<?> clazz : clazzs) {
            Method[] methods = clazz.getDeclaredMethods();
            Object bean = SpringUtil.getBean(clazz);
            NatsListener listener = clazz.getDeclaredAnnotation(NatsListener.class);
            if (listener.value() == clientType && methods.length > 0) {
                Dispatcher dispatcher = nc.createDispatcher();
                for (Method method : methods) {
                    if (!method.isAnnotationPresent(Subscribe.class)) {
                        continue;
                    }
                    Subscribe subscribe = method.getAnnotation(Subscribe.class);
                    method.setAccessible(true);
                    if (subscribe.mode() == Mode.SIMPLE) {
                        subscribe(method, bean , dispatcher);
                    } else {
                        jetStreamSubscribe(method, bean);
                    }
                }
            }
        }
        log.info("[{}] finished subscribe topic", this.getClientType());
    }


    @Override
    public Message request(Message message, long ms) {
        Message msg = null;
        try {
            msg = nc.requestWithTimeout(message, Duration.ofMillis(ms)).get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        return msg;
    }

    private void consumerClear() throws IOException {
        if (!consumerMap.isEmpty()) {
            JetStreamManagement js = nc.jetStreamManagement();
            consumerMap.forEach((durable, method) -> {
                try {
                    Subscribe subscribe = method.getDeclaredAnnotation(Subscribe.class);
                    String streamName = subscribe.streamName();
                    ConsumerInfo consumerInfo = js.getConsumerInfo(streamName, durable);
                    if (consumerInfo != null) {
                        js.deleteConsumer(streamName, durable);
                    }
                } catch (IOException | JetStreamApiException ignored) {
                }
            });
        }
        consumerMap.clear();
    }


    /**
     * @return 获取客户端类型
     */
    @Override
    public ClientType getClientType() {
        return this.clientType;
    }


    /**
     * 发送请求
     *
     * @param message nats message
     * @param timeout 超时时间
     * @return 返回结果
     */
    @Override
    public CompletableFuture<Message> requestWithTimeout(Message message, long timeout) {
        return nc.requestWithTimeout(message , Duration.ofMillis(timeout));
    }

    @Override
    public void destroy() {
        try {
            if (this.nc != null) {
                consumerClear();
                if ((this.nc.getStatus() != Connection.Status.CLOSED)){
                    this.nc.drain(drainTimeout).get();
                }else {
                    this.nc.close();
                }
                this.nc = null;
                log.info("[{}] | nats destroy finished", this.getClientType());
            }
        } catch (Exception e) {
            log.error("[{}] | nats fail to destroy", this.getClientType(), e);
        }
    }
}
